package org.tigr.htc.server;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.LineNumberReader;
import java.io.Reader;
import java.io.StringReader;
import java.net.InetAddress;
import java.net.MalformedURLException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;
import org.apache.xmlrpc.WebServer;
import org.apache.xmlrpc.XmlRpcClient;
import org.tigr.htc.cmd.CmdState;
import org.tigr.htc.cmd.Command;
import org.tigr.htc.cmd.CommandHelper;
import org.tigr.htc.cmd.CommandSetRoot;
import org.tigr.htc.cmd.CommandType;
import org.tigr.htc.cmd.ICommand;
import org.tigr.htc.cmd.Task;
import org.tigr.htc.cmd.TaskGroup;
import org.tigr.htc.common.HTCConfig;
import org.tigr.htc.persist.IPersist;
import org.tigr.htc.persist.PersistFactory;

/**
 * The <b>CommandServer</b> class supplies methods that allows users to submit distributed
 * jobs through XML-RPC. This class is a singleton.
 *
 * To get the single instance of this class, call
 * the static getInstance() method.  For example:
 * <pre>
 *     CommandServer cserver = CommandServer.getInstance();
 * </pre>
 *
 * CommandServer depends on XML-RPC and Castor
 */

public class CommandServer {
	static Logger logger = Logger.getLogger(CommandServer.class);
	/**
	 * The instance variable <code>cserver</code> holds the single instance of the server class
	 */
	static private CommandServer cserver = null;
	
	/**
	 * The variable <code>sched</code> holds the the instance of the scheduler class used to
	 * schedule submissions.
	 */
	static private Scheduler sched = null;
	/**
	 * The variable <code>HOSTIP</code> stores the IP address of the server.
	 */
	static public String HOSTIP = null;
	/**
	 * The variable <code>hostName</code> holds the host name of this server
	 */
	private static String hostName = null;
    
	/**
	 * The variable <code>pl</code> holds the class used to persist the object to the datastore
	 */
	static public IPersist pl = PersistFactory.getIPersist();
	
	/**
	 * The variable <code>peerList</code> keeps the list of peer servers that are active
	 */
	static private List peerList = Collections.synchronizedList(new Vector());
    
    // client XML-RPC object to send procedure calls to server
    static private XmlRpcClient client;
    
    static Map peerClients = new HashMap();
    
	/**
	 *  Private constructor so more than one cserver server can be created
	 *
	 */
	private CommandServer() {
		try {
			// Get the IP address of the server and store it for later use
			HOSTIP = InetAddress.getLocalHost().getHostAddress();
			// TODO: Remove hard coded domain name...
			hostName = InetAddress.getLocalHost().getHostName() + ".tigr.org";
			logger.info("Server IP address is " + HOSTIP + ". Hostname (FQDN): " + hostName + ".");
			
			// Create an instance of the scheduler and launch the thread that schedules the jobs
			sched = Scheduler.getInstance();
			Thread t = new Thread(sched);
			logger.info("Starting scheduler...");
			t.start();
			logger.info("Scheduler started ");
			
            // Create a thread that prints the memory amount every minute
            Thread memThread = new Thread () {
                public void run () {
                    Runtime runtime = Runtime.getRuntime();
                    
                    // Run this thread forever
                    while (true) {
                        logger.info("Max memory: " + runtime.maxMemory() + " total memory: "
                                + runtime.totalMemory() + " free memory: " + runtime.freeMemory());
                        try {
                            Thread.sleep(60000);
                        }
                        catch (InterruptedException e) {
                        } // Sleep for a second
                    }
                }
            };
            memThread.start();
            
            logger.info("Creating peer list.");
			try {
                // Open the file that lists the peers and add to the peer list
                String peerFile = HTCConfig.getPeerList();
                logger.debug("Peer list file name is " + peerFile);
                LineNumberReader lnr = new LineNumberReader(new FileReader(
                        peerFile));
                String peerName = null;
                while ((peerName = lnr.readLine()) != null) {
                    logger.info("Read peer name " + peerName);
                    if (!peerName.equals(hostName)) {
                        peerList.add(peerName);
                        // Create the client needed to communicate with peer and add to map 
                        String uri = "http://" + peerName + ":"
                                + HTCConfig.getProperty("htc.service.port")
                                + HTCConfig.getProperty("htc.service.address");
                        logger.info("Creating RPC client for peer using URI: "
                                + uri);
                        client = new XmlRpcClient(uri);
                        peerClients.put(peerName, client);
                    }
                }
                lnr.close();
            } catch (FileNotFoundException e) {
                logger.warn("Caught FileNotFoundException while creating the peer list");
                e.printStackTrace();
            } catch (IOException e) {
                logger.warn("Caught IOException while creating the peer list");
                e.printStackTrace();
            } catch (RuntimeException e) {
                logger.warn("Caught RuntimeException while creating the peer list");
                e.printStackTrace();
            }
			logger.info("Created peer list.");
            
			// Spawn a set of monitors that will monitor the jobs
			logger.info("Creating monitor threads.");
			
		    // Get the number of monitor threads from the config file
			int monitorThreads = Integer.parseInt(HTCConfig.getProperty("MONITOR_THREAD_COUNT"));
			for (int i=0; i<monitorThreads; i++) {
				RunnerMonitor.spawnMonitor();
			}
			
            logger.info("Created monitor threads.");
			
            // Before adding the commands for re-monitoring hold submissions so the MAXRUNNING
            // can be honored
            sched.setHoldSubmits(true);
            
            logger.info("Identifying commands that need to be remonitored.");
			// Get all the active commands that were being monitored by this server by reading all the commands
			// which are running, waiting, or in init state and being monitored by this host
            List activeCommands = pl.searchCommands(" monitorHost=\"" + hostName + "\" " +
                    " and ( (state = \"INIT\") or (state = \"WAITING\") or (state = \"RUNNING\"))" +
                    " and (type = \"htc\")", null, null);

			// If nothing is found to monitor then set to empty Vector
			activeCommands = (activeCommands == null) ? new Vector() : activeCommands;
            try {
                for (Iterator iter = activeCommands.iterator(); iter.hasNext();) {
                    Command cmd = (Command) iter.next();

                    // Check to see if this command had an event listener registered
                    // in which case re-register the event listener. If we do this it
                    // will send all the events including the ones that might have been
                    // sent earlier, but we will have to deal with it at a later date

                    logger.debug("Command is " + cmd);
                    String url = cmd.getSubmitURL();
                    logger.debug("URL is " + url);
                    
                    // In the past we were only checking the URL, now whenever the user adds a listener
                    // we set the send events flag to true, so check that flag as well as URL
                    // before deciding to add a listener
                    if (cmd.isSendEvents() && (url != null)) {
                        RequestProxy rp = new RequestProxy(url);
                        cmd.addStatusListener(rp);
                        cmd.addRuntimeStatusListener(rp);
                    }

                    // Register this command for re-monitoring
                    sched.remonitorCommand(cmd);
                }
            } catch (Exception e) {
                // In case there was an error re-monitoring this command write warning message
    			logger.warn("Error in remonitoring the existing active jobs -- " + e.getMessage(), e);
            }
            logger.info("Finished remonitoring initilization.");
            
            // After all the previous commands have been either queued or added to monitor queue
            // release the hold on submissions
            sched.setHoldSubmits(false);
		} catch (Exception e) {
			logger.fatal("Command server constructor: " + e.getMessage(), e);
			System.exit(1);
		}
	}
	
	/**
	 * CommandServer is a singleton and doesn't have a public constructor 
	 *
	 */
	public synchronized static CommandServer getInstance() {
		if (cserver == null) {
			logger.debug("cserver was null, it will be created...");
			cserver = new CommandServer();
		}
		return cserver; 
	}
		
	/**
	 * The method <code>createCmdID</code> creates a new command ID and returns the ID
	 * 
	 * @return a <code>String</code> representing the created command ID
	 */
	public String createCmdID() {
		long id = Command.createCmdID();
		
		logger.info("Giving out new id " + id);
		return Long.toString(id);
	}
	
	/**
	 * The method <code>getCommand</code> returns the command associated with the specified
	 * ID. This allows a client to connect and get the latest command data
	 * 
	 * @param ID a <code>String</code> representation of the command ID
	 * 
	 * @return a <code>String</code> representation of the command object
	 */
	public String getCommand(String ID) {
		logger.debug("In getCommand. ID: " + ID);
		long id = Long.parseLong(ID);
		
		// Locate the command in the datastore and rebuild the command object
		Command cmd = (Command) pl.find(id, Command.class);
		logger.debug("Exit HTCServer.getCommand. Got Command " + cmd.getID());
		
		return cmd.toString();
	}
    
    /**
     * The method <code>getTasks</code> returns the string representation of the tasks
     * of the array job that was launched by the specified request ID. If this request is made
     * on a non-array command then an empty string is returned.
     * 
     * @param pID a <code>String</code> representing the request ID
     * 
     * @return a <code>String</code> representation of the command tasks and their status,
     * an empty string if the command could not be found, or if the wrong command type was used
     */
    public String getTasks(String pID) {
        if (logger.isDebugEnabled()) {
            logger.debug("In getTasks for ID: " + pID);
        }
        
        // First locate the command and make sure it is of the MW type
        long id = Long.parseLong(pID);
        Command cmd = (Command) pl.find(id, Command.class);
        if (cmd == null) {
            logger.warn("Command not found.");
            return "Command not found.";
        }
        if (cmd.getType().equals(CommandType.MW)) {
            // As this type is an MW runner type, build the task groups
            Vector tasks = (Vector) pl.retrieveTasks(id);
            TaskGroup tg = new TaskGroup();
            tg.setCmdId(id);
            tg.setTasks(tasks);
            
            // Go through each of the tasks and set the array parameter
            for (int i = 0; i < tasks.size(); i++) {
                Task tsk = (Task) tasks.get(i);
                try {
                    tsk.setParam(cmd.getParamAtIndex(i));
                }
                catch (IllegalStateException e) {
                	// ignored
                }
                catch (IOException e) {
                	// ignored
                }
            }
            return tg.toXMLString();
        } else {
			logger.warn("Invalid Command type encountered: " + cmd.getType());
			return "";
        }
    }
    
    /**
     * The method <code>monitorHost</code> returns the name of the HTC worker monitoring the job 
     * with the specified ID. 
     * 
     * @param pID a <code>String</code> representation of the command ID
     * @return a <code>String</code> representation of the host name
     */
    public String monitorHost(String pID) {
        logger.debug("In monitorHost: " + pID);
        long id = Long.parseLong(pID);
        
        // Locate the command in the datastore and rebuild the Command object
        Command cmd = (Command) pl.find(id, Command.class);
        
        return cmd.getMonitorHost();
    }
    
	/**
	 * The method <code>addListener</code> method registers the specified URL as the
	 * RPC service as interested in listening to events associated with the specified ID.
	 * <p>
	 * The method first checks to see if a runner for this ID exists in the set of runners
	 * on this server. If it can find a runner then it registers status and runtime listener
	 * to the command associated with the runner.
	 * <p>
	 * If the runner cannot be found then return an error. This can be a problem in two instances,
	 * if the request to listen goes to a server that is not the one to which the original submit
	 * event was sent, or if between the point of submit and a request to register a listener the
	 * job was complete on the underlying grid. Both of these are problems and need to be fixed.
	 * 
	 * 
	 * @param url a <code>String</code> representing the URL of the client to which the proxy
	 * can send events.
	 * @param pId a <code>String</code> representing the command ID
	 * @param type a <code>String</code> representing the type of command, currently not used
	 * @return
	 */
	public int addListener(String url, String pId, int type) {
		logger.debug("Enter HTCServer.addListener " + pId + " " + url);
		try {
			long id = Long.parseLong(pId);
			
			// Locate the runner that is chaperoning this command and register the
            // listener, if a runner cannot be found then handle the request differently
            // by either checking to see the job may have completed, or forwarding to other
            // peer servers if they are monitoring
			IRunner runner = sched.findRunner(id);
			if (runner != null) {
				ICommand cmd = runner.getICommand();
				logger.debug("Found runner and adding request proxy");
				
				// Create a new request proxy with the client URL and register
				// it as the listener for the command
				RequestProxy rp = new RequestProxy(url);
				cmd.addStatusListener(rp);
				cmd.addRuntimeStatusListener(rp);
                
                // As the user has requested events back change the status of send events flag and update
                Command htcCommand = (Command) cmd;
                htcCommand.setSendEvents(true);
                htcCommand.updateCommand();
			}
			else {
				logger.debug("No Runner found.");
				logger.debug("Number of peers is: " + peerList.size());
			    // As a runner does not exist first recreate the command
			    // by reading from the persistence layer
				ICommand cmd = (ICommand) pl.find(id, Command.class);
				if (cmd.getState().equals(CmdState.FAILED) ||
                        cmd.getState().equals(CmdState.FINISHED) ||
                        cmd.getState().equals(CmdState.INTERRUPTED)) {
					logger.debug("Command is either failed, finished or interrupted.");
				    // See if the command is no longer running because it 
				    // completed or failed between the time it was
				    // submitted and the request to add listener in which case send the
				    // event back immediately                    
                    forceEvents(url, id, cmd);
                    
				}
				else if (peerList.size() > 0) {
					logger.debug("Number of peers is " + peerList.size());
				    // As there is no active runner and the command was not already complete
				    // check with peers if they know anything about this job
                    for (Iterator iter = peerList.iterator(); iter.hasNext();) {
                        String peer = (String) iter.next();
                        logger.debug("Attempting to forward request to peer " + peer);
                        client = (XmlRpcClient) peerClients.get(peer);
                        try {
                            Vector params = new Vector();
                            params.add(url); // Call back URI
                            params.add(Long.toString(id)); // Command ID
                            params.add("0"); // Type
                            Integer i = (Integer) client.execute("HTCServer.addListenerForPeer", params);
                            int retVal = i.intValue();
                            if (retVal == 1) {
                                // This peer is listening so having forwarded request return
                                if (logger.isDebugEnabled()) {
                                    logger.debug("Forwarded addListener request to peer.");
                                }
                                return retVal;
                            }
                        } catch (Exception e) {
                            logger.debug("Request monitor error ", e);
                            return -1;
                        }
                    }
                    
                    // After trying to forward the request check the status again in case
                    // between the requests the job completed before returning with an error
                    cmd = (ICommand) pl.find(id, Command.class);
                    if (cmd.getState().equals(CmdState.FAILED) ||
                            cmd.getState().equals(CmdState.FINISHED) ||
                            cmd.getState().equals(CmdState.INTERRUPTED)) {
                        // See if the command is no longer running because it 
                        // completed or failed between the time it was
                        // submitted and the request to add listener in which case send the
                        // event back immediately
                        forceEvents(url, id, cmd);
                    }
				}
				else {
					logger.error("Request to addlistener to non-existent command " + pId);
					return -1;
				}		    
			}
		} catch (MalformedURLException e) {
			logger.error("Bad url passed to addlistener:" + e.getMessage());
			return -1;
		}
        catch (Exception e) {
            logger.error("Could not create runner:" + e.getMessage());
            return -1;
        }
		
		return 1;
	}

    /**
     * The method <code>forceEvents</code> forces all the events generated for this command to be resent
     * to the client host. This is useful in instances where the job finished before a listener could
     * be registered.
     * 
     * @param url a <code>String</code> representing the URL of the client to which the proxy
     * can send events.
     * @param id a <code>long</code> representing the command ID
     * @param cmd a <code>ICommand</code> representing the command
     * 
     * @return a <code>boolean</code> of true if the events could be sent back, false otherwise
     * 
     * @throws MalformedURLException
     * @throws Exception a <code></code>
     */
    private boolean forceEvents(String url, long id, ICommand cmd) throws MalformedURLException, Exception {
        logger.info("Forcing events for command with ID: " + id);
        
        // Create a new request proxy with the client URL and register
        // it as the listener for the command
        Command command = (Command) cmd;
        RequestProxy rp = new RequestProxy(url);
        cmd.addStatusListener(rp);
        cmd.addRuntimeStatusListener(rp);
        
        // Create a new runner and call the monitor method on it
        // so all the vents could be sent in the order they were received
        File parentDir = Scheduler.getParentDir(id);
        if (logger.isDebugEnabled()) {
            logger.debug("Create runner.");
        }
        IRunner runner = RunnerFactory.newRunner(command, parentDir);
        return runner.isRunnerComplete();
    }

    /**
     * The method <code>addListener</code> registers the specified URL as the
     * RPC service as interested in listening to events associated with the specified ID.
     * <p>
     * This method is meant to be invoked by other peer servers to forward requests and hence only 
     * check to see if a runner for this command exists and registers the listener with this
     * runner, otherwise it returns
     * 
     * 
     * @param url a <code>String</code> representing the URL of the client to which the proxy
     * can send events.
     * @param pId a <code>String</code> representing the command ID
     * @param type a <code>String</code> representing the type of command, currently not used
     * 
     * @return an <code>int</code> value of 1 if the listener could be registered successfully,
     * a value of less than 0 otherwise
     */
    public int addListenerForPeer(String url, String pId, String type) {
        if (logger.isDebugEnabled()) {
            logger.debug("Enter HTCServer.addListenerForPeer " + pId + " " + url + " peer "
                    + hostName);
        }
        try {
            long id = Long.parseLong(pId);
            
            // Locate the runner that is chaperoning this command and register the
            // listener, if a runner cannot be found then handle the request differently
            // by either checking to see the job may have completed, or forwarding to other
            // peer servers if they are monitoring
            IRunner runner = sched.findRunner(id);
            if (runner != null) {
                ICommand cmd = runner.getICommand();
                logger.debug("Found runner and adding request proxy.");
                
                // Create a new request proxy with the client URL and register
                // it as the listener for the command
                RequestProxy rp = new RequestProxy(url);
                cmd.addStatusListener(rp);
                cmd.addRuntimeStatusListener(rp);
                
                // As the user has requested events back change the status of send events flag and update
                Command htcCommand = (Command) cmd;
                htcCommand.setSendEvents(true);
                htcCommand.updateCommand();
            } else {
                if (logger.isDebugEnabled()) {
                    logger.debug("Could not locate runner.");
                }
                return 0;
            }
        } catch (MalformedURLException e) {
            logger.error("Bad url passed to addlistener:" + e.getMessage());
            return -1;
        }
        
        return 1;
    }
    
	/**
	 * The method <code>submit</code> submits the request to the underlying grid technology
     * using the environment specified.
	 * 
	 * @param command a <code>String</code> representation of the command to invoke
	 * @param env a <code>Vector</code> representing the environment under which the
	 * command should run.
	 * 
	 * @return a <code>String</code> representing the command ID to indicate success,
     * otherwise a message representing the error.
	 */
	public String submit(String command, HashMap env) {
		StringReader fr = null;
		ICommand cmd = null;
		
		try {
			if(logger.isDebugEnabled()) {
				logger.debug("----- Start XML request ----");
				logger.debug(command);
				logger.debug("----- End XML request ----");
			}
			
			// Convert XML-RPC string to stream 
			fr = new StringReader(command);
			
			// Unmarshal the XML into Java objects
			CommandSetRoot root = CommandHelper.getInstance().unmarshalRoot((Reader)fr);
			cmd = root.getCommand();
			((Command) cmd).setMonitorHost(hostName); // Set the monitor host name for the command to be used later
		} catch (Exception e) {
			logger.warn("Could not validate the XML request string");
			logger.warn(e.getMessage());
			return e.getMessage();
		} finally {
			if (fr != null) {
				fr.close();
			}
		}
		
		try {
			cmd.setEnv(env);
			if(cmd.getID() < 1) {
				logger.error("Command had an id of zero!!! Resetting id.");
				cmd.setID(Command.createCmdID());
			}
			
            // Add the command to the queue
			logger.info("About to enqueue Command " + cmd.getID());
			sched.enqueue(cmd);
			
			return Long.toString(cmd.getID());
		} catch(Exception e) {
			logger.error("Could not enqueue Command " + cmd.getID());
			logger.error(e.getMessage(), e);
			return e.getMessage();
		}
	}
    
    /**
     * The method <code>submit</code> submits the request to the underlying grid technology
     * using no environment.
     * 
     * @param command a <code>String</code> representation of the command to invoke
     * 
     * @return a <code>String</code> representing the Command ID to indicate success,
     * otherwise a message representing the error.
     */	
	public String submit(String command) {
		HashMap env = null;
		return submit(command, env);
	}
    
    /**
     * The method <code>submit</code> submits the request to the underlying grid technology
     * using no environment. This method in addition to submitting the job also
     * registers a request proxy to send events to the client as the command changes state.
     * 
     * @param command a <code>String</code> representation of the command to invoke
     * @param pUri a <code>String</code> representing the client URI to send notifications
     * 
     * @return a <code>String</code> representing the command ID to indicate success,
     * otherwise a message representing the error.
     */	
	public String submit(String command, String pUri) {
		HashMap env = null;
		return submit(command, pUri, env);
	}
    
    /**
     * The method <code>submit</code> submits the request to the underlying grid technology
     * using the environment specified. This method in addition to submitting the job also
     * registers a request proxy to send events to the client as the command changes state.
     * 
     * @param command a <code>String</code> representation of the command to invoke
     * @param pURI a <code>String</code> representing the client URI to send notifications
     * @param env a <code>Vector</code> representing the environment under which the
     * command should run.
     * 
     * @return a <code>String</code> representing the command ID to indicate success,
     * otherwise a message representing the error.
     */
	public String submit(String command, String pURI, HashMap env) {
		StringReader fr = null;
		ICommand cmd;
		
		try {
			if (logger.isDebugEnabled()) {
				logger.debug("----- Start XML request ----");
				logger.debug(command);
				logger.debug("----- End XML request ----");
				logger.debug("Client at URL: " + pURI);
			}
			
			// Convert XML-RPC string to stream 
			fr = new StringReader(command);
			
			// Unmarshal XML into Java objects
			CommandSetRoot root = CommandHelper.getInstance().unmarshalRoot((Reader)fr);
			cmd = root.getCommand();
		} catch(Exception e) {
			logger.warn("Could not validate XML request string");
			logger.warn(e.getMessage());
			return "0";
		} finally {
			if(fr != null) {
				fr.close();
			}
		}
		
		try {
            // Set the monitor host value for this Command
            Command htcCommand = (Command) cmd;
            htcCommand.setMonitorHost(hostName);
            htcCommand.setSendEvents(true); // As the user has registered a URI, set sendEvents flag
            
			cmd.setEnv(env);
			if(cmd.getID() < 1) {
				logger.error("Command had an ID of zero!!! Resetting id...");
				cmd.setID(Command.createCmdID());
			}
            
			// client proxy stuff 
			RequestProxy rp = new RequestProxy(pURI);
			cmd.addRuntimeStatusListener(rp);
			cmd.addStatusListener(rp);
			
			// Enqueue command
			logger.debug("About to enqueue cmd " + cmd.getID());
			sched.enqueue(cmd);
			
			// TODO
			// For disconnect and reconnect
			// clients.put(new Long(cmd.getID()), rp);
			
			return Long.toString(cmd.getID());
		} catch(Exception e) {
			logger.error("Could not enqueue cmd " + cmd.getID());
			logger.error(e.getMessage());
			return e.getMessage();
		}
	}
	
	// check file exists on server and is on an NFS mount 
	// using 'df' doesn't seem good, need to use something like stat
	public boolean isNFS(String path) throws FileNotFoundException {
		File testFile = new File(path);
		Process p;
		
		for (int i=0; i < 2; i++) {
			if (testFile.exists()) {
				try {
                    // This invocation of df only works on the linux platform. It will not work
                    // on Solaris for example. There, it would be called like so: df -F nfs
					p = Runtime.getRuntime().exec("/bin/df -t nfs " + path);
					BufferedReader out = new BufferedReader(new InputStreamReader(p.getInputStream()));
					try {
						String line = null;
						
						line = out.readLine();  // eat header
						line = out.readLine();
						if (line != null) {
							return true;
						}
						return false;
					} finally {
						out.close();
					}
				} catch (Exception e) {
					logger.debug("Exception " + e);
					return false;
				}
			} else {
				if ((testFile = testFile.getParentFile()) != null) {
					path = testFile.getAbsolutePath();
				} else {
					break;
				}
			}
		}
		return false;
	}

	/**
	 * The method <code>status</code> returns the status of the command specified with the 
     * command ID
	 * 
	 * @param pID a <code>String</code> representing the command ID
     * 
	 * @return a <code>Hashtable</code> containing the results. The hash has two keys, the 'ID'
     * which represents the command ID, and the 'state' which holds the current state
	 */
	public Hashtable status(String pID) {
		logger.debug("In status with id: " + pID);
		Hashtable result = null;
		long id = Long.parseLong(pID);

		ICommand cmd = (ICommand) pl.find(id, Command.class);
		
		if (cmd != null) {
			String state = cmd.getState().toString();
			logger.debug("Found job " + pID + " in a state of  " + state);
			result = new Hashtable();
			
			result.put("ID", pID);
			result.put("state", state);
			
			return result;
		} else {
			logger.debug("Job " + pID + " not in cache. Get it from the database.");
		}
		
		return result;
	}
	
    /**
     * The method <code>status</code> returns the status of the command specified with the 
     * command ID
     * 
     * @param id an <code>int</code> representing the command ID
     * 
     * @return a <code>Hashtable</code> containing the results. The hash has two keys, the 'ID'
     * which represents the command ID, and the 'state' which holds the current state
     */
	public Hashtable status(int id) {
		logger.debug("In status: " + id);
		return status(Integer.toString(id));
	}
	
	/**
	 * The method <code>stop</code> stops the command execution for the Command with the specified
     * ID. The method launches a separate thread to actually execute the stop. The method does not
     * actually check if the job was stopped on the grid before returning, it only attempts to stop 
     * the job.
	 * 
	 * @param id a <code>String</code> representing the Command ID
	 * 
     * @return a <code>boolean</code> of true if the thread to stop job was launched successfully,
     * else return false
	 */
	public boolean stop(String id) {
		try {
			logger.debug("In stop: " + id);
			
			Thread MyThread = new Thread() { // Create an anonymous Thread subclass
				public void run() { 
					CommandServer.getInstance().realStop(this.getName());
				}
			};
			
			MyThread.setName(id);
			MyThread.start();
			
			logger.debug("Leaving stop: " + id);
			return true;
		} catch(Exception e) {
			logger.warn("Could not stop job ID.", e);
			return false;
		}
	}

    /**
     * The method <code>isMonitoring</code> checks to see if this server is monitoring the
     * request with the specified ID
     * 
     * @param id a <code>String</code> representing the command ID
     * 
     * @return a <code>boolean</code> of true if this command is being monitored, false otherwise
     */
    public boolean isMonitoring(String id) {
        try {
            logger.debug("In isMonitoring: " + id);

            long cid = Long.parseLong(id);
            IRunner runner = sched.findRunner(cid);
            if (runner != null) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Monitoring Command.");
                }
                return true;
            }
        } catch (Exception e) {
            logger.warn("Could not determine if monitoring job.", e);
            return false;
        }
        if (logger.isDebugEnabled()) {
            logger.debug("Not monitoring command.");
        }
        return false;
    }
    
	/**
	 * The method <code>realStop</code> removes the job from the GRID
	 * 
     * @param id a <code>String</code> representing the Command ID
     * 
	 * @return a <code>boolean</code> of true if the jobs was stopped successfully, false
     * otherwise
	 */
	public boolean realStop(String id) {
		logger.debug("In realStop: " + id);
		
		try {
			long cid = Long.parseLong(id);
			IRunner runner = sched.findRunner(cid);
			if(runner != null) {
				logger.debug("Found runner. Sending stop request for " + id);
				runner.stop();
			}
            else if (peerList.size() > 0) {
                // As there is no active runner and the command was not already complete
                // check with peers if they know anything about this job
                for (Iterator iter = peerList.iterator(); iter.hasNext();) {
                    String peer = (String) iter.next();
                    logger.debug("Attempting to forward request to peer " + peer);
                    client = (XmlRpcClient) peerClients.get(peer);
                    try {
                        Vector params = new Vector();
                        params.add(id); // Command ID
                        Boolean bool = (Boolean) client.execute("HTCServer.isMonitoring", params);
                        if (bool == Boolean.TRUE) {
                            // This peer is monitoring so forwarded stop request
                            if (logger.isDebugEnabled()) {
                                logger.debug("Forwarding stop request to peer.");
                            }
                            client.execute("HTCServer.stop", params);
                            return bool.booleanValue();
                        }
                    } catch (Exception e) {
                        logger.debug("Request monitor error ", e);
                        return false;
                    }
                }
            } else {
                // As the command is not being monitored here or by any of the peers it must have stopped
                // so return
                if (logger.isDebugEnabled()) {
                    logger.debug("Stop could not be processed as the Command seems to have completed.");
                }
            }
			logger.debug("In stop: " + id);
			return true;
		} catch(Exception e) {
			logger.warn("Could not stop job id.", e);
			return false;
		}
		
	}
	
	/**
	 * For testing only, not for general use
	 */
	public static void main(String args[]) {
		int PORT = 8080;
		int MWPORT = 9090;
		
		try {
			PropertyConfigurator.configureAndWatch("conf/log4j_server.conf");
			cserver = CommandServer.getInstance();
						
			logger.debug("Create server");
			WebServer webserver = new WebServer(PORT);
			
			logger.debug("Add handler");
			webserver.addHandler("HTCServer", cserver);
			
			logger.debug("Start");
			webserver.start();
			
			logger.debug("Create MW Server");
			
			logger.debug("Add handler");
			
			logger.debug("Start");
			
		} catch (Exception e) {
			logger.fatal("main -- " + e.getMessage(), e);
		}
	}
}
